/*-----------------------------------------------------------------------------
 * 
 *
 *-----------------------------------------------------------------------------
 */

#include "server/include/stock_server_queue.h"
#include <pthread.h>

#define STOCK_MAX_QUEUE (16*1024)

static stock_update    stock_server_queue[STOCK_MAX_QUEUE];
static pthread_mutex_t stock_queue_mutex;
static pthread_cond_t  stock_queue_cond;
static int             read_ptr = 0;
static int             write_ptr = 0;
static int             queue_size = 0;


void
stock_queue_dump()
{
    printf("\nStock Queue Dump:\n");
    printf("  read_ptr:   %d\n", read_ptr);
    printf("  write_ptr:  %d\n", write_ptr);
    printf("  queue_size: %d\n\n", queue_size);
}


static int
next_read()
{
    read_ptr = ((read_ptr + 1) % STOCK_MAX_QUEUE);
    return read_ptr; 
}


static int
next_write() 
{
    write_ptr =  ((write_ptr + 1) % STOCK_MAX_QUEUE);
    return write_ptr;
}


void
stock_server_queue_init()
{
    memset(stock_server_queue, 0, sizeof(stock_update) * STOCK_MAX_QUEUE);
    pthread_mutex_init(&stock_queue_mutex, NULL);
    pthread_cond_init(&stock_queue_cond, NULL);
}


void 
stock_server_enqueue(stock_update *update)
{
    pthread_mutex_lock(&stock_queue_mutex);
    memcpy(&stock_server_queue[next_write()], update, sizeof(stock_update));
    queue_size++;
    pthread_cond_broadcast(&stock_queue_cond);
    pthread_mutex_unlock(&stock_queue_mutex);
}


stock_update * 
stock_server_dequeue()
{
    stock_update *update = NULL;;

    pthread_mutex_lock(&stock_queue_mutex);

    while (queue_size == 0) {
        pthread_cond_wait(&stock_queue_cond, &stock_queue_mutex);
    }

    update = &stock_server_queue[next_read()];
    queue_size--;
    pthread_mutex_unlock(&stock_queue_mutex);

    return update;
}

